package com.caul.demo.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * Created by sdliang on 2018/4/24.
 */
public class SparkDemo {

  private static final String URI_HDFS = "hdfs://cdh-namenode:9000";

  /**
   * 加入"-DHADOOP_USER_NAME=root",可以伪装当前用户为root
   *
   * @param args
   */
  public static void main(String[] args) {
    //    wordCount(args);
    //    collectionMerge(args);
    popularTeacher(args);
  }

  private static List<String> recordList = new ArrayList<>();

  static {
    recordList.add("http://java.study.com/wangqiang");
    recordList.add("http://java.study.com/liming");
    recordList.add("http://java.study.com/wangqiang");
    recordList.add("http://java.study.com/wangqiang");
    recordList.add("http://java.study.com/liming");
    recordList.add("http://java.study.com/wangqiang");
    recordList.add("http://java.study.com/liming");
    recordList.add("http://java.study.com/wangqiang");

    recordList.add("http://php.study.com/wujie");
    recordList.add("http://php.study.com/wujie");
    recordList.add("http://php.study.com/wujie");

    recordList.add("http://golang.study.com/zhangliang");
    recordList.add("http://golang.study.com/lilei");
    recordList.add("http://golang.study.com/zhangliang");
    recordList.add("http://golang.study.com/zhangliang");

  }

  private static void popularTeacher(String[] args) {
    SparkConf conf = new SparkConf().setAppName("popularTeacher").setMaster("local");
    JavaSparkContext context = new JavaSparkContext(conf);
    JavaRDD<String> javaRDD = context.parallelize(recordList);
    //先转成<URI,1>
    JavaPairRDD<String, Integer> langRDD = javaRDD.mapToPair(str -> Tuple2.apply(str, 1));
    //按URI分组统计<URI,count>
    JavaPairRDD<String, Integer> strContRDD = langRDD.reduceByKey(Integer::sum);
    //转成<Integer, URI>, 倒叙排序, 再转回来<URI,count>
    JavaPairRDD<String, Integer> sortedRDD = strContRDD.mapToPair(Tuple2::swap).sortByKey(false)
        .mapToPair(Tuple2::swap);
    //转成<language,Teacher>
    JavaPairRDD<String, Teacher> teacherRDD = sortedRDD.mapToPair(tuple -> Tuple2
        .apply(Teacher.getLangName(tuple._1), Teacher.apply(tuple._1, tuple._2)));
    //按language分组
    JavaPairRDD<String, Iterable<Teacher>> resultRDD = teacherRDD.groupByKey();
    //输出
    for (Tuple2<String, Iterable<Teacher>> tuple : resultRDD.collect()) {
      System.out.println("------------------statistic language------------------" + tuple._1);
      tuple._2.forEach(System.out::println);
    }
  }


  private static void collectionMerge(String[] args) {
    SparkConf conf = new SparkConf().setAppName("collectionMerge").setMaster("local");
    JavaSparkContext context = new JavaSparkContext(conf);

    List<Tuple2<String, String>> leftList = new ArrayList<>();
    leftList.add(Tuple2.apply("1", "张三"));
    leftList.add(Tuple2.apply("2", "李四"));
    leftList.add(Tuple2.apply("3", "王五"));
    JavaRDD<Tuple2<String, String>> jRDDLeft = context.parallelize(leftList);

    List<Tuple2<String, String>> rightList = new ArrayList<>();
    leftList.add(Tuple2.apply("1", "张三"));
    leftList.add(Tuple2.apply("3", "宝强"));
    leftList.add(Tuple2.apply("4", "小六"));
    JavaRDD<Tuple2<String, String>> jRDDRight = context.parallelize(rightList);

    JavaRDD<Tuple2<String, String>> union = jRDDLeft.union(jRDDRight);
    List<Tuple2<String, String>> collect = union.collect();
    System.out.println("pring union..." + collect.size());
    for (Tuple2<String, String> user : collect) {
      System.out.println(user);
    }

    JavaRDD<Tuple2<String, String>> intersection = jRDDLeft.intersection(jRDDRight);
    collect = intersection.collect();
    System.out.println("pring intersection..." + collect.size());
    for (Tuple2<String, String> user : collect) {
      System.out.println(user);
    }

    JavaRDD<Tuple2<String, String>> subtract = jRDDLeft.subtract(jRDDRight);
    collect = subtract.collect();
    System.out.println("pring subtract..." + collect.size());
    for (Tuple2<String, String> user : collect) {
      System.out.println(user);
    }

    JavaPairRDD<Tuple2<String, String>, Tuple2<String, String>> cartesian = jRDDLeft
        .cartesian(jRDDRight);
    List<Tuple2<Tuple2<String, String>, Tuple2<String, String>>> collect2 = cartesian.collect();
    System.out.println("pring cartesian..." + collect2.size());
    for (Tuple2<Tuple2<String, String>, Tuple2<String, String>> tuple : collect2) {
      System.out.println(tuple._1);
      System.out.println(tuple._2);
    }
  }

  /**
   * 统计单词数
   *
   * @param args
   */
  private static void wordCount(String[] args) {
    SparkConf conf = new SparkConf().setAppName("WordCount").setMaster("local");
    //    conf.set("", "root");
    JavaSparkContext context = new JavaSparkContext(conf);

    //读取源文件
    JavaRDD<String> rdd = context.textFile(URI_HDFS + args[0]);
    //对一条记录进行逗号切分, 并整合
    JavaRDD<String> words = rdd.flatMap(str -> Arrays.asList(str.split(",")).iterator());
    //将word转成<word,1>
    JavaPairRDD<String, Integer> wordsCont = words.mapToPair(str -> Tuple2.apply(str, 1));
    //聚合word,并根据Key统计数量
    JavaPairRDD<String, Integer> wordsSum = wordsCont.reduceByKey(Integer::sum);
    //调换key-value,后排序,再调换顺序
    JavaPairRDD<String, Integer> swapPairs = wordsSum.mapToPair(Tuple2::swap).sortByKey(false)
        .mapToPair(Tuple2::swap);
    //输出(注意: 如果save到本地,需要先collect,否则本地没数据, 数据只在worker节点上)
    swapPairs.saveAsTextFile(URI_HDFS + args[1]);

    //释放资源
    context.stop();
  }

  private static class Teacher implements Serializable {

    private String language;
    private String name;
    private Integer count;

    public Teacher() {
    }

    public Teacher(String language, String name, Integer count) {
      this.language = language;
      this.name = name;
      this.count = count;
    }

    public String getLanguage() {
      return language;
    }

    public void setLanguage(String language) {
      this.language = language;
    }

    public String getName() {
      return name;
    }

    public void setName(String name) {
      this.name = name;
    }

    public Integer getCount() {
      return count;
    }

    public void setCount(Integer count) {
      this.count = count;
    }

    @Override
    public String toString() {
      return "language:" + language + ";name:" + name + ";count:" + count;
    }

    public static Teacher apply(String str, Integer count) {
      Teacher teacher = new Teacher();
      teacher.setCount(count);
      teacher.setLanguage(getLangName(str));
      teacher.setName(getName(str));
      return teacher;
    }

    public static String getLangName(String str) {
      return str.substring(7, str.indexOf("."));
    }

    private static String getName(String str) {
      return str.substring(str.lastIndexOf("/") + 1);
    }
  }

}
